package com.greate.community.config;

import com.greate.community.util.CommunityConstant;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.errors.TopicExistsException;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

@Configuration
@EnableKafka
public class KafkaConfig implements CommunityConstant {

    // 可以改为你自己的 Kafka 地址
    private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:9092";
    @PostConstruct
    public void createKafkaTopics() {
        Properties props = new Properties();
        props.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS); // Kafka broker 地址

        // 创建 AdminClient 实例
        try (AdminClient adminClient = AdminClient.create(props)) {
            List<NewTopic> topics = Arrays.asList(
                    new NewTopic(TOPIC_COMMNET, 3, (short) 2),
                    new NewTopic(TOPIC_LIKE, 3, (short) 2),
                    new NewTopic(TOPIC_FOLLOW, 3, (short) 2),
                    new NewTopic(TOPIC_PUBLISH, 3, (short) 2),
                    new NewTopic(TOPIC_DELETE, 3, (short) 2)
            );

            try {
                // 批量创建 Kafka 主题
                KafkaFuture<Void> future = adminClient.createTopics(topics).all();
                future.get(); // 等待创建完成
                System.out.println("Topics created successfully!");
            } catch (TopicExistsException e) {
                System.out.println("Some topics already exist: " + e.getMessage());
            } catch (Exception e) {
                e.printStackTrace();
                System.err.println("Error occurred while creating topics: " + e.getMessage());
            }
        }
    }
    // 可选：如果你希望通过 Spring 管理 Kafka 生产者和消费者，可以在这里定义 KafkaTemplate 等 Bean

//    @Bean
//    public KafkaTemplate<String, String> kafkaTemplate() {
//        // KafkaProducer 配置，使用 Spring Kafka 管理生产者
//        Properties properties = new Properties();
//        properties.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
//        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//
//        DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(properties);
//        return new KafkaTemplate<>(factory);
//    }
}
